package pattern.socketimpl;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;
import pattern.cache.InitCache;
import pattern.dto.User;
import pattern.enums.dict.work.UserTypeEnum;
import pattern.feign.WorkServiceFeign;
import pattern.result.ResponseData;
import pattern.result.SocketResult;
import pattern.utils.CollectionUtil;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;

/**
 * 发送socket消息管理
 * @author Simon
 * @todo 对没有发送成功用户未做处理
 */

@Component
@Slf4j
@RequiredArgsConstructor(onConstructor_ = @Autowired)
public class SocketSendProxy {

    private final InitCache initCache;

    private final SimpMessagingTemplate simpMessagingTemplate;

    private final WorkServiceFeign workServiceFeign;

    public SimpMessagingTemplate getSimpMessagingTemplate() {
        return simpMessagingTemplate;
    }

    /**
     * 对应订阅主题发送消息
     */
    public <T> void sendMsgByTopic(String topic, ResponseData<T> responseData) {
        simpMessagingTemplate.convertAndSend(topic, responseData);
    }

    /**
     * 点对点发送消息
     */
    public <T> void toOneByToken(String token, String topic, ResponseData<T> responseData) {
        simpMessagingTemplate.convertAndSendToUser(token, topic, responseData);
    }

    /**
     * 点对点发送消息
     */
    public <T> SocketResult toOneByUserId(Collection<Integer> userIds, String topic, ResponseData<T> responseData) {
        SocketResult socketResult = new SocketResult();
        for (Integer userId : userIds) {
            var sessionIds = initCache.socketSessionUserId.get(userId);
            if (CollectionUtil.isEmpty(sessionIds)) {
                socketResult.getFailUsers().add(userId);
                continue;
            }
            for (String sessionId : sessionIds) {
                simpMessagingTemplate.convertAndSendToUser(sessionId, topic, responseData,createHeaders(sessionId));
            }
            socketResult.getSuccessUsers().add(userId);
        }
        socketResult.exec();
        return socketResult;
    }

    /**
     * UUID点对点发送消息
     */
    public <T> SocketResult toOneByUUID(String uuid, String topic, ResponseData<T> responseData) {
        SocketResult socketResult = new SocketResult();
        String sessionId = initCache.socketSessionUUID.get(uuid);
        if (StringUtils.isBlank(sessionId)) {
            return new SocketResult();
        }
        simpMessagingTemplate.convertAndSendToUser(uuid, topic, responseData,createHeaders(sessionId));
        socketResult.exec();
        return socketResult;
    }

    /**
     * 点对多发送消息demo
     * @param userTypeEnum：用户类型
     * @param topic：主题
     * @param responseData：消息内容
     */
    public <T> SocketResult toMany(UserTypeEnum userTypeEnum, String topic, ResponseData<T> responseData) {
        SocketResult socketResult = new SocketResult();
        List<User> users = workServiceFeign.searchUserByType(userTypeEnum.getCode());
        try {
            //用户过多采用多线程发送
            List<List<User>> lists = CollectionUtil.split(users, 200);
            ScheduledExecutorService downPool = new ScheduledThreadPoolExecutor(lists.size(),
                    new BasicThreadFactory.Builder().namingPattern("sendSocketToMany-pool-%d").daemon(Boolean.TRUE).build());
            for (List<User> user : lists) {
                Future<SocketResult> submit = downPool.submit(() -> toOneByUserId(CollectionUtil.convertList(user,User::getWorkerId), topic, responseData));
                socketResult.getSuccessUsers().addAll(submit.get().getSuccessUsers());
                socketResult.getFailUsers().addAll(submit.get().getFailUsers());
            }
            downPool.shutdown();
            downPool.awaitTermination(1, TimeUnit.HOURS);
        } catch (InterruptedException | ExecutionException e) {
            log.error("发送socket消息", e);
        }
        socketResult.exec();
        return socketResult;
    }

    private MessageHeaders createHeaders(String sessionId) {
        SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
        headerAccessor.setSessionId(sessionId);
        headerAccessor.setLeaveMutable(true);
        return headerAccessor.getMessageHeaders();
    }

}
